package com.shujia.gma.ads

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.types.Row

object Demo1ProvinceIndex {

  def main(args: Array[String]): Unit = {
    /**
      * 指标：
      * 1.每天每个省份的订单数
      *  2.每天每个省份的订单金额
      *
      */

    val fsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner() //计划器
      .inStreamingMode()
      .build()

    val fsEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    fsEnv.setParallelism(1)
    val fsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(fsEnv, fsSettings)


    fsTableEnv.executeSql(
      """
        |
        |
        |-- ---------------------------------
        |-- DWD层,支付订单明细表dwd_paid_order_detail
        |-- ---------------------------------
        |CREATE TABLE dwd_paid_order_detail
        |(
        |  detail_id BIGINT,
        |  order_id BIGINT,
        |  user_id BIGINT,
        |  province_id INT,
        |  sku_id BIGINT,
        |  sku_name STRING,
        |  sku_num INT,
        |  order_price DECIMAL(10,0),
        |  create_time TIMESTAMP(0),
        |  pay_time TIMESTAMP(0)
        | ) WITH (
        |    'connector' = 'kafka',
        |    'topic' = 'dwd.paid_order_detail',
        |    'scan.startup.mode' = 'earliest-offset',
        |    'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
        |    'format' = 'changelog-json'
        |)
        |
      """.stripMargin)

    fsTableEnv.executeSql(
      """
        |CREATE TABLE dim_base_region (
        | id int,
        | i ROW<name STRING,region_name STRING,area_code STRING>,
        | PRIMARY KEY (id) NOT ENFORCED
        |) WITH (
        | 'connector' = 'hbase-1.4',
        | 'table-name' = 'dim:base_region',
        | 'zookeeper.quorum' = 'master:2181,node1,2181,node2:2181',
        | 'sink.buffer-flush.max-rows'='1'
        |)
        |
      """.stripMargin)

    /**
      *
      * 先创建ads数据库
      * 先到mysql厚葬出啊关键结果表
      *
      *
CREATE TABLE ads.ads_province_index(
  province_id INT(10),
  area_code VARCHAR(100),
  province_name VARCHAR(100),
  region_name VARCHAR(100),
  order_amount DECIMAL(10,2),
  order_count BIGINT(10),
  dt VARCHAR(100),
  PRIMARY KEY (province_id, dt)
) ;
      *
      */


    fsTableEnv.executeSql(
      """
        |
        |
        |-- ---------------------------------
        |-- 使用 DDL创建MySQL中的ADS层表
        |-- 指标：1.每天每个省份的订单数
        |--      2.每天每个省份的订单金额
        |-- ---------------------------------
        |CREATE TABLE ads_province_index(
        |  province_id INT,
        |  area_code STRING,
        |  province_name STRING,
        |  region_name STRING,
        |  order_amount DECIMAL(10,2),
        |  order_count BIGINT,
        |  dt STRING,
        |  PRIMARY KEY (province_id, dt) NOT ENFORCED
        |) WITH (
        |    'connector' = 'jdbc',
        |    'url' = 'jdbc:mysql://master:3306/ads?useUnicode=true&characterEncoding=UTF-8',
        |    'table-name' = 'ads_province_index',
        |    'driver' = 'com.mysql.jdbc.Driver',
        |    'username' = 'root',
        |    'password' = '123456'
        |)
        |
        |
      """.stripMargin)


    fsTableEnv.executeSql(
      """
        |insert into ads_province_index
        |select
        |a.province_id,
        |b.i.area_code as area_code,
        |b.i.name as province_name,
        |b.i.region_name as region_name,
        |a.order_amount as order_amount,
        |a.order_count as order_count,
        |a.dt as dt
        |from
        |(select
        | province_id,
        | DATE_FORMAT(pay_time,'yyyy-MM-dd') as dt,
        | count(distinct order_id) as order_count,
        | sum(sku_num*order_price) as order_amount
        | from
        |dwd_paid_order_detail
        |group by province_id,DATE_FORMAT(pay_time,'yyyy-MM-dd')
        |) as a
        |join
        |dim_base_region as b
        |on a.province_id=b.id
        |
        |
      """.stripMargin)


  }

}
